package com.demo.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.demo.kafka.constant.TopicConstant;
import com.demo.kafka.entity.KafkaMessage;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * 测试Kafka发送消息
 * @author cyj
 */
@Slf4j
@Service
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, KafkaMessage> kafkaTemplate;

    /**
     * 发送消息
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void send(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send(TopicConstant.TOPIC_1, kafkaMessage).get();
        log.info("result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送消息并指定分配
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void sendPartition0(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send(TopicConstant.TOPIC_1, 0, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendPartition0 result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送消息并指定分配
     * @param kafkaMessage 消息
     */
    @SneakyThrows
    public void sendPartition1(KafkaMessage kafkaMessage) {
        Object object = kafkaTemplate.send(TopicConstant.TOPIC_1, 1, kafkaMessage.getMsgId(), kafkaMessage).get();
        log.info("sendPartition1 result：{}", JSON.toJSONString(object));
    }

    /**
     * 发送多条消息，测试消费批量消息
     * @param kafkaMessage 消息
     */
    public void send2(KafkaMessage kafkaMessage) {
        kafkaTemplate.send(TopicConstant.TOPIC_2, kafkaMessage);
    }
}
